package com.test.flink_source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * kafka source的本质是消费者
 * 消费者在消费数据的时候为了保证数据的一致性会记录偏移量
 *
 */
public class F05KafkaSource {

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8888);
        StreamExecutionEnvironment webUI = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("127.0.0.1:9092,127.0.0.1:9092,127.0.0.1:9092")
                .setGroupId("edit-41")
                .setTopics("hero")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        DataStreamSource<String> ds = webUI.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kfk");

        ds.print();
        webUI.execute();
    }
}
